package com.cjfather.redis.steam.service;

import java.time.Duration;
import java.util.Map;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.stereotype.Service;

import com.cjfather.redis.steam.controller.PropellingInfo;

import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;

/**
 * 
 * @author: todd
 * @date:   2023-08-08
 */


@Service
@Slf4j
public class CommonService {
	
	@Resource(name="redisMq6Factory")
    private RedisTemplate redisMq6Factory;

	@Qualifier("reactiveRedisObjectTemplate")
	@Autowired
    private ReactiveRedisTemplate<String, Object> redisTemplate;

    private StreamOperations<String, String, String> stream = null;
    
    @Value("${cjfather.mq}")
    String mq;
    
    @PostConstruct
    public void init(){
        LettuceConnectionFactory connectionFactory = (LettuceConnectionFactory) redisMq6Factory.getConnectionFactory();
        // 不启用共享连接 允许多个连接公用一个物理连接。如果设置false ,每一个连接的操作都会开启和关闭socket连接。如果设置为false,会导致性能下降
        connectionFactory.setShareNativeConnection(false);
        stream = redisMq6Factory.opsForStream();
    }
    
    public String addStream(String qkey,Map<String,String> data) {
    	JSONArray mqs = JSONUtil.parseArray(mq);
    	//此处打开可以防止组被误删除，但性能会下降2/3   	
    	if(!redisMq6Factory.hasKey(qkey)) {
    		for(JSONObject keys:mqs.toList(JSONObject.class)) {
        		String key = keys.getStr("key");
        		
        		for(JSONObject group:keys.getJSONArray("groups").toList(JSONObject.class)) {
        	    	String gname = group.getStr("group");	
        	    	
                    try {
                    	redisMq6Factory.opsForStream().createGroup(key,ReadOffset.from("0-0"), gname);
        			} catch (Exception e) {
        				log.info("{}组存在",gname);
        			}
        		}
        	}
    	}
    	RecordId recordId = stream.add(qkey, data);
    	return recordId.getValue();
    }
    
    
    public String addStream(String qkey,JSONObject data) {
    	JSONArray mqs = JSONUtil.parseArray(mq);
//此处打开可以防止组被误删除，但性能会下降2/3   	
    	if(!redisMq6Factory.hasKey(qkey)) {
    		for(JSONObject keys:mqs.toList(JSONObject.class)) {
        		String key = keys.getStr("key");
        		
        		for(JSONObject group:keys.getJSONArray("groups").toList(JSONObject.class)) {
        	    	String gname = group.getStr("group");	
        	    	
                    try {
                    	redisMq6Factory.opsForStream().createGroup(key,ReadOffset.from("0-0"), gname);
        			} catch (Exception e) {
        				log.info("{}组存在",gname);
        			}
        		}
        	}
    	}
    	ObjectRecord<String, JSONObject> stringUserObjectRecord = ObjectRecord.create(qkey, data);
    	RecordId recordId = stream.add(stringUserObjectRecord);
    	return recordId.getValue();
    }
    
    public String addStream(String qkey,String data) {
    	JSONArray mqs = JSONUtil.parseArray(mq);
//此处打开可以防止组被误删除，但性能会下降2/3   	
    	if(!redisMq6Factory.hasKey(qkey)) {
    		for(JSONObject keys:mqs.toList(JSONObject.class)) {
        		String key = keys.getStr("key");
        		
        		for(JSONObject group:keys.getJSONArray("groups").toList(JSONObject.class)) {
        	    	String gname = group.getStr("group");	
        	    	
                    try {
                    	redisMq6Factory.opsForStream().createGroup(key,ReadOffset.from("0-0"), gname);
        			} catch (Exception e) {
        				log.info("{}组存在",gname);
        			}
        		}
        	}
    	}
    	ObjectRecord<String, String> stringUserObjectRecord = ObjectRecord.create(qkey, data);
    	RecordId recordId = stream.add(stringUserObjectRecord);
    	return recordId.getValue();
    }
    
    public String addStreamMap(String qkey,Map data) {
    	JSONArray mqs = JSONUtil.parseArray(mq);
    	//此处打开可以防止组被误删除，但性能会下降2/3   	
    	if(!redisMq6Factory.hasKey(qkey)) {
    		for(JSONObject keys:mqs.toList(JSONObject.class)) {
        		String key = keys.getStr("key");
        		
        		for(JSONObject group:keys.getJSONArray("groups").toList(JSONObject.class)) {
        	    	String gname = group.getStr("group");	
        	    	
                    try {
                    	redisMq6Factory.opsForStream().createGroup(key,ReadOffset.from("0-0"), gname);
        			} catch (Exception e) {
        				log.info("{}组存在",gname);
        			}
        		}
        	}
    	}
    	ObjectRecord<String, Map> stringUserObjectRecord = StreamRecords.newRecord().in(qkey).ofObject(data);
    	RecordId recordId = stream.add(stringUserObjectRecord);
    	return recordId.getValue();
    }
    
    public String addStreamObj(String qkey,Object data) {
    	ObjectRecord<String, Object> stringUserObjectRecord = StreamRecords.newRecord().in(qkey).ofObject(data);
    	RecordId recordId = redisTemplate.opsForStream().add(stringUserObjectRecord).block();
		Mono<Boolean> setResult = redisTemplate.opsForValue().set("stream1", recordId);

		Boolean setSuccess = setResult.block(Duration.ofSeconds(10));
		System.out.println("设置结果: " + setSuccess);

		if (Boolean.TRUE.equals(setSuccess)) {
			// 2. 立即获取
			Mono<Object> mono = redisTemplate.opsForValue().get("stream1");

			mono.subscribe(
					value -> System.out.println("成功: " + value),           // onNext
					error -> System.out.println("错误: " + error.getMessage()), // onError
					() -> System.out.println("操作完成")                     // onComplete
			);

//			// 添加调试信息
//			mono = mono
//					.doOnSubscribe(s -> System.out.println("开始获取值"))
//					.doOnNext(v -> System.out.println("获取到值: " + v))
//					.doOnSuccess(v -> System.out.println("获取成功: " + v))
//					.doOnError(e -> System.err.println("获取失败: " + e.getMessage()));

			String value = mono.block() + "";
			System.out.println("最终结果: " + value);
		}

    	return recordId.getValue();
    }


}
